package function;

import bean.orderWide;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import redis.clients.jedis.Jedis;
import utils.RedisUtil;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class dimMapFunction extends RichMapFunction<orderWide,orderWide> {
    Connection connection=null;
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        connection = DriverManager.getConnection("jdbc:phoenix:192.168.1.10:2181");
    }

    @Override
    public void close() throws Exception {
        if (connection==null){
            connection.close();
        }
    }

    @Override
    public orderWide map(orderWide value) throws Exception {
        //查询Phoenix之前先查询Redis
        Jedis jedis = RedisUtil.getJedis();
        String key = "dim_test"+value.getOrderwideId();
        String jvalue = jedis.get(key);
        if(jvalue!=null){
            jedis.expire(key,24 * 60 * 60);
            jedis.close();
            JSONObject jsonObject = JSONObject.parseObject(jvalue);
            String jsonString = jsonObject.toJSONString();
            orderWide orderWide = JSONObject.parseObject(jsonString, orderWide.class);
            System.out.println("从redis读取数据");
            return orderWide;
        }else{
            jedis.close();
        }
        String orderId = value.getOrderwideId();
        StringBuffer sql = new StringBuffer();
        sql.append("select city,country from ns1.test where id ='"+orderId+"'");
        //预编译sql
        PreparedStatement preparedStatement = connection.prepareStatement(sql.toString());
        ResultSet resultSet = preparedStatement.executeQuery();
        while(resultSet.next()){
            value.setCity(resultSet.getString("city"));
            value.setCountry(resultSet.getString("country"));
        }
        jedis.set(key,JSONObject.toJSONString(value));
        return value;
    }
}
